package consumer;

import base.ConsumerMsgDispatcher;
import base.MQBase;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;

import java.util.Map;

/**
 * @author : zGame
 * @version V1.0
 * @Project: rokectmqprac
 * @Package consumer
 * @Description: pullConsumer 已经不推荐使用
 * @date Date : 2022年01月11日 11:04
 */
public abstract class ConsumerUtil extends MQBase {
    
    protected DefaultMQPushConsumer consumer;
    private base.ConsumerMsgDispatcher ConsumerMsgDispatcher;
    
    public ConsumerUtil() {
        try {
            consumer = new DefaultMQPushConsumer(getGroup());
            propertySet();
            //配置消息监听
            consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
                try {
                    msgs.forEach(msg -> {
                        ConsumerMsgDispatcher.dispatch(msg);
                    });
                    
                } catch (Exception e) {
                    System.err.println("接收异常" + e);
                }
                return ConsumeOrderlyStatus.SUCCESS;
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
